#include "Connection.h"
#include "Channel.h"
#include "EventLoop.h"

Connection::Connection(EventLoop *loop, Socket clientSock)
        : loop(loop), clientSock(std::move(clientSock)), disconnect(false),
          clientChannel(new Channel(loop, clientSock.fd))
{
    clientChannel->onRead = [this]() { OnMessage(); };
    clientChannel->onWrite = [this]() { Write(); };
    clientChannel->onError = [this]() { Error(); };
    clientChannel->onClose = [this]() { Close(); };
    clientChannel->UseEt();
    clientChannel->EnableReading();
}

Connection::~Connection()
{
    SAFE_CLOSESOCKET(clientSock.fd);
}

int Connection::GetFd() const
{
    return clientSock.fd;
}

std::string Connection::GetIp() const
{
    return clientSock.ip;
}

uint16_t Connection::GetPort() const
{
    return clientSock.port;
}

void Connection::OnMessage()
{
    char buffer[1024];
    while (true) {
        bzero(&buffer, sizeof(buffer));
        ssize_t readNum = read(GetFd(), buffer, sizeof(buffer));
        if (readNum > 0) {
            inputBuffer.Append(buffer, readNum);
        } else if (readNum == -1 && errno == EINTR) {
            continue;
        } else if (readNum == -1 && ((errno == EAGAIN) || (errno == EWOULDBLOCK))) {
            std::string message;
            while (true) {
                if (!inputBuffer.PickMessage(message)) {
                    break;
                }
                lastTime = time(nullptr);
                onMessage(shared_from_this(), message);
            }
            break;
        } else if(readNum == 0) {
            Close();
            break;
        }
    }
}

void Connection::Close()
{
    disconnect = true;
    clientChannel->Remove();
    onClose(shared_from_this());
}

void Connection::Error()
{
    disconnect = true;
    clientChannel->Remove();
    onError(shared_from_this());
}

void Connection::Write()
{
    ssize_t writen = ::send(GetFd(), outputBuffer.Data(), outputBuffer.Size(), 0);
    if (writen > 0) {
        outputBuffer.Erase(0, writen);
    }
    if (outputBuffer.Size() == 0) {
        clientChannel->DisableWriting();
        onSendComplete(shared_from_this());
    }
}

void Connection::Send(const char *data, size_t size)
{
    if (disconnect) {
        return;
    }
    std::shared_ptr<std::string> message(new std::string(data));
    if (loop->IsInLoopThread()) {
        SendInLoop(message);
    } else {
        loop->QueueInLoop([&] { SendInLoop(message); });
    }
}

void Connection::SendInLoop(const std::shared_ptr<std::string>& data)
{
    outputBuffer.AppendWithSep(data->data(), data->size());
    clientChannel->EnableWriting();
}

bool Connection::Timeout(time_t now, int val) const
{
    return now - lastTime > val;
}
